package org.com.blbl.sortHotTopicTop5;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.com.blbl.bean.BiliSortBean;
import org.com.blbl.sortHighQualityUpTop10.SortHighQualityUpTop10Driver;
import org.com.blbl.sortHighQualityUpTop10.SortHighQualityUpTop10Mapper;
import org.com.blbl.sortHighQualityUpTop10.SortHighQualityUpTop10Reducer;

import java.io.IOException;

public class SortHotTopicTop5Driver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        //设置连接客户端的身份。
        //System.setProperty("HADOOP_USER_NAME", "root");
        //创建驱动类
        Configuration conf = new Configuration();
        //设置文件系统 默认当前程序所在的文件路径
        //因为 我们的要求是从HDFS中读取数据文件 以及 分析的结果 也要放到HDFS中
        //conf.set("fs.defaultFS", "hdfs://node1:8020");

        //构造job作业的实例,参数（配置对象，job名字）
        Job job = Job.getInstance(conf, SortHotTopicTop5Driver.class.getSimpleName());

        //设置mr程序运行的主类
        job.setJarByClass(SortHotTopicTop5Driver.class);

        //设置本次mr程序的mapper类型、reducer类型
        job.setMapperClass(SortHotTopicTop5Mapper.class);
        job.setReducerClass(SortHotTopicTop5Reducer.class);

        //指定mapper阶段输出的key value数据类型
        job.setMapOutputKeyClass(BiliSortBean.class);
        job.setMapOutputValueClass(Text.class);

        //指定reducer阶段输出的key value数据类型，也是mr程序最终的输出数据类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(BiliSortBean.class);

        //配置本次作业的输入数据路径和输出数据路径
        Path inputPath = new Path("output/HotTopic");
        Path outputPath = new Path("output/SortHotTopicTop5");

        //todo 默认组件 TextInputFormat TextOutputFormat
        FileInputFormat.setInputPaths(job, inputPath);
        FileOutputFormat.setOutputPath(job, outputPath);

        //判断输出路径是否已经存在，如果存在了删掉
        FileSystem fs = FileSystem.get(conf);
        if (fs.exists(outputPath)) {
            fs.delete(outputPath, true);//递归删除
        }
        //最终提交本次作业
        boolean resultFlag = job.waitForCompletion(true);
        //退出程序 和 job进行绑定 。0正常退出 1是异常退出
        System.exit(resultFlag ? 0 : 1);
    }
}
